package com.hksj.app.websocket;


import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hksj.common.config.utils.SpringContextUtil;
import com.hksj.common.core.model.Rest;
import com.hksj.app.websocket.message.RequestMessage;
import com.hksj.app.websocket.message.ResponseMessage;
import com.hksj.app.websocket.message.Subscription;
import com.hksj.jetlinks.manager.TokenManager;
import com.hksj.jetlinks.model.LoginUser;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

@Component
@ServerEndpoint("/ws/stream")
public class ScreenWebsocket {
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    /**
     * 每隔多久推一次消息
     */
    private Integer interval = 5;

    /**
     * 0 = 待连接 1 = 已连接 2 = 已关闭
     */
    private Integer status = 0;

    /**
     * 订阅的接口列表
     */
    private List<Subscription> subscriptionList = new ArrayList<>();

    private ObjectMapper objectMapper;

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        this.status = 1;
        ((ThreadPoolTaskExecutor) SpringContextUtil.getBean(ThreadPoolTaskExecutor.class)).execute(() -> init());
        System.out.println("连接成功 sessionId:" + session.getId() + this);
        session.getAsyncRemote().sendText(Rest.ok("连接成功").toString());
    }


    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        this.status = 2;
        this.subscriptionList.clear();
        System.out.println("关闭成功 sessionId:" + session.getId());
    }

    /**
     * 发生错误时调用
     */
    @OnError
    public void onError(Session session, Throwable error) {
        error.printStackTrace();
    }


    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        try {
            RequestMessage requestMessage = JSONObject.parseObject(message, RequestMessage.class);
            if (!ObjectUtils.isEmpty(requestMessage) && Objects.equals(requestMessage.getType(),"subscribe")) {
                TokenManager tokenManager = SpringContextUtil.getBean(TokenManager.class);
                LoginUser userInfo = tokenManager.getUserInfo(requestMessage.getToken());
                if (ObjectUtils.isEmpty(userInfo)){
                    sendMessage(Rest.fail("认证失败").toString());
                    return;
                }
                sendMessage(Rest.ok("订阅成功",null).toString());
                if (!ObjectUtils.isEmpty(requestMessage.getSubscriptionList())){
                    this.interval = requestMessage.getInterval();
                    this.subscriptionList = requestMessage.getSubscriptionList();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void sendMessage(ResponseMessage message){
        if (ObjectUtils.isEmpty(objectMapper)){
            objectMapper = SpringContextUtil.getBean(ObjectMapper.class);
        }
        try {
            sendMessage(objectMapper.writeValueAsString(message));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    /**
     * 向链接推送消息
     * @param message
     * @throws IOException
     */
    public void sendMessage(String message){
        this.session.getAsyncRemote().sendText(message);
    }

    public void init() {
        try{
            while (true){
                if (status == 2){
                    break;
                }
                if (!ObjectUtils.isEmpty(subscriptionList)){
                    for (Subscription subscription : subscriptionList) {
                        Object data = ((ApiAdapter)SpringContextUtil.getBean(ApiAdapter.class)).invoke(subscription);
                        sendMessage(new ResponseMessage(subscription.getUrl(),data));
                    }
                }
                //限制不小于3秒 保护系统
                Thread.sleep((interval < 3 ? 3 : interval) * 1000);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
